package com.unitd.modules.configcenter.component;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.unitd.frame.comm.utils.DateUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import com.unitd.frame.comm.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @desc  自定义组件,用于定义配置信息的状态持有者信息
 * @filename ConfigStateHolder.java
 * @copyright www.unitd.com
 * @author Hudan
 * @version 1.0
 * @date  2018-01-30
 */
@Component
public class ConfigStateHolder implements InitializingBean {

	private static Logger logger = LoggerFactory.getLogger(ConfigStateHolder.class.getName());

	private static final String NOTIFY_UPLOAD_CMD = "upload";
	/** 配置中心信息存储在zookeeper服务上的默认根目录 */
	private final static String ZK_ROOT_PATH = "/confcenter";
	/** 当前Spring的运行环境 */
	private @Autowired
	Environment environment;
	/** 配置对象信息 */
	private static Map<String, List<ConfigState>> configStates = new ConcurrentHashMap<>();
	/** 同步类型 */
	public static final String SYNC_TYPE_ZK = "zookeeper";
	/** zookeeper客户端 */
	private static ZkClient zkClient;

	/**
	 * @desc 实现InitializingBean接口, 重写读取properties属性文件后的处理方法
	 */
	@Override
	public void afterPropertiesSet() {
		try {
			String zkServers = environment.getProperty("zookeeper.servers");
			if (StringUtils.isNotBlank(zkServers)) {
				ZkConnection zkConnection = new ZkConnection(zkServers);
				zkClient = new ZkClient(zkConnection, 10000);
				logger.info("config_zookeeper.servers {} init finish", zkServers);

				// 如果配置信息不存在, 则同步配置信息
				syncConfigIfMiss();
			}
		} catch (Exception e) {
			zkClient = null;
			logger.error("init config_zookeeper.servers error", e);
		}
	}

	/**
	 * @desc 如果配置信息不存在, 则同步配置信息
	 */
	private void syncConfigIfMiss() {
		if (!zkClient.exists(ZK_ROOT_PATH)) {
			return;
		}

		List<String> envs = zkClient.getChildren(ZK_ROOT_PATH);

		String parentPath;
		List<String> apps;
		for (String env : envs) {
			parentPath = ZK_ROOT_PATH + "/" + env;
			apps = zkClient.getChildren(parentPath);
			for (String app : apps) {
				parentPath = ZK_ROOT_PATH + "/" + env + "/" + app + "/nodes";

				int activeNodeCount = zkClient.countChildren(parentPath);
				if (activeNodeCount == 0) continue;
				List<ConfigState> localCacheConfigs = configStates.get(app + "#" + env);

				if (activeNodeCount > 0 && (localCacheConfigs == null || localCacheConfigs.size() < activeNodeCount)) {
					zkClient.writeData(parentPath, NOTIFY_UPLOAD_CMD);
					logger.info("send cmd[{}] on path[{}]", NOTIFY_UPLOAD_CMD, parentPath);
				}
			}
		}
	}

	/**
	 * @desc 添加属性配置信息
	 * @param cs 属性配置信息
	 */
	public void add(ConfigState cs) {
		String key = cs.getAppName() + "#" + cs.getEnv();
		List<ConfigState> list = configStates.get(key);
		if (list == null) {
			list = new ArrayList<>();
			configStates.put(key, list);
			list.add(cs);
		} else {
			if (!list.contains(cs)) {
				list.add(cs);
			}
		}
		logger.info("New node[{}-{}-{}] registered!", cs.getAppName(), cs.getEnv(), cs.getNodeId());
	}

	/**
	 * @desc 根据key获取属性配置信息
	 * @param env
	 * @return
	 */
	public List<ConfigState> get(String env) {

		List<ConfigState> result = new ArrayList<>();
		Set<String> keys = configStates.keySet();

		for (String key : keys) {
			if (key.endsWith(env)) {
				clearExpireNodes(key);
				List<ConfigState> list = configStates.get(key);
				result.addAll(list);
			}
		}
		return result;
	}

	public List<ConfigState> get(String appName, String env) {
		List<ConfigState> clist = configStates.get(appName + "#" + env);
		return clist == null ? new ArrayList<>() : clist;
	}

	/**
	 * @desc 清除过期的服务节点
	 * @param appNameAnaEnvKey 属性key
	 */
	private static void clearExpireNodes(String appNameAnaEnvKey) {
		try {
			List<ConfigState> sameAppNodes = configStates.get(appNameAnaEnvKey);
			if (sameAppNodes == null || sameAppNodes.isEmpty()) return;
			String syncType = sameAppNodes.get(0).syncType;
			String zkPath = sameAppNodes.get(0).zkPath;

			Date nowTime = new Date();
			if (SYNC_TYPE_ZK.equals(syncType)) {
				List<String> activeNodes = zkClient.getChildren(zkPath);
				if (activeNodes == null) return;
				if (activeNodes.isEmpty()) {
					sameAppNodes.clear();
					return;
				}
				Iterator<ConfigState> iterator = sameAppNodes.iterator();
				while (iterator.hasNext()) {
					ConfigState c = iterator.next();
					if (!activeNodes.contains(c.getNodeId())) {
						logger.info("remove expire node:{}", c.getNodeId());
						iterator.remove();
					} else {
						c.setSyncTime(nowTime);
					}
				}
			} else {
				synchronized (sameAppNodes) {
					Iterator<ConfigState> iter = sameAppNodes.iterator();
					while (iter.hasNext()) {
						ConfigState s = iter.next();
						if (DateUtils.getDiffSeconds(nowTime, s.syncTime) > s.syncIntervalSeconds * 2) {
							iter.remove();
						}
					}
				}
			}
		} catch (Exception e) {
			logger.error("clearExpireNodes:" + appNameAnaEnvKey, e);
		}
	}

	/**
	 * @desc 自定义的配置中心属性状态信息类
	 */
	public static class ConfigState {
		/** 节点ID */
		private String nodeId;
		/** 应用名称 */
		private String appName;
		/** 运行环境 */
		private String env;
		/** 版本号 */
		private String version;
		/** 是否是springboot环境 */
		private boolean springboot;
		/** 配置中心的服务端IP */
		private String serverip;
		/** 配置中心的服务端端口 */
		private int serverport;
		/** 配置中心客户端与服务端进行心跳的间隔时间(单位:秒) */
		private int syncIntervalSeconds = 90;
		/** 配置中心客户端与服务端进行通讯的方式,默认:HTTP */
		private String syncType = "http";
		/*** 同步时间 */
		private Date syncTime = new Date();
		/** 同步的zookeeper数据目录地址 */
		private String zkPath;

		/**
		 * 所有配置属性信息 */
		@JsonIgnore
		private Map<String, String> configs = new TreeMap<>();
		/** 等待进行同步的配置信息 */
		private Map<String, String> waitingSyncConfigs = new HashMap<>();

		/**
		 * @desc 构造方法,用于初始化一些属性的基本信息
		 * @param allMaps
		 */
		public ConfigState(Map<String, String> allMaps) {
			this.nodeId = allMaps.remove("nodeId");
			this.appName = allMaps.remove("appName");
			this.env = allMaps.remove("env");
			this.version = allMaps.remove("version");
			if (allMaps.containsKey("syncType")) syncType = allMaps.remove("syncType");
			if (allMaps.containsKey("syncIntervalSeconds"))
				this.syncIntervalSeconds = Integer.parseInt(allMaps.remove("syncIntervalSeconds"));
			springboot = Boolean.parseBoolean(allMaps.remove("springboot"));
			if (allMaps.containsKey("serverport")) {
				serverport = Integer.parseInt(allMaps.remove("serverport"));
			}
			if (allMaps.containsKey("serverip")) {
				serverip = allMaps.remove("serverip");
			}

			ArrayList<String> keys = new ArrayList<>(allMaps.keySet());
			Collections.sort(keys);
			for (String key : keys) {
				configs.put(key, allMaps.get(key));
			}

			zkPath = ZK_ROOT_PATH + "/" + env + "/" + appName + "/nodes";
			if (SYNC_TYPE_ZK.equals(syncType)) {
				if (zkClient == null) {
					logger.warn("Zookeeper client not init,skip");
					return;
				}

				if (!zkClient.exists(zkPath)) return;
				clearExpireNodes(appName + "#" + env);
				logger.info("int ConfigState ok");
			}
		}

		public void update(String nodeId, Map<String, String> datas) {
			if (datas.containsKey("serverip")) {
				serverip = datas.get("serverip");
			}
			syncTime = new Date();
			// 清理失效节点
			//clearExpireNodes(appName + "#" + env);
		}

		public void onConfigSyncSuccess() {
			if (waitingSyncConfigs.isEmpty()) return;
			configs.putAll(waitingSyncConfigs);
			waitingSyncConfigs.clear();
		}

		public String getNodeId() {
			return nodeId;
		}
		public void setNodeId(String nodeId) {
			this.nodeId = nodeId;
		}
		public String getAppName() {
			return appName;
		}
		public void setAppName(String appName) {
			this.appName = appName;
		}
		public String getEnv() {
			return env;
		}
		public void setEnv(String env) {
			this.env = env;
		}
		public String getVersion() {
			return version;
		}
		public void setVersion(String version) {
			this.version = version;
		}
		public Map<String, String> getConfigs() {
			return configs;
		}
		public void setConfigs(Map<String, String> configs) {
			this.configs = configs;
		}
		public Date getSyncTime() {
			return syncTime;
		}
		public void setSyncTime(Date syncTime) {
			this.syncTime = syncTime;
		}
		public boolean isSpringboot() {
			return springboot;
		}
		public String getServerip() {
			return serverip;
		}
		public void setServerip(String serverip) {
			this.serverip = serverip;
		}
		public int getServerport() {
			return serverport;
		}
		public void setServerport(int serverport) {
			this.serverport = serverport;
		}
		public void setSpringboot(boolean springboot) {
			this.springboot = springboot;
		}
		public Map<String, String> getWaitingSyncConfigs() {
			return waitingSyncConfigs;
		}
		public String getSyncType() {
			return syncType;
		}

		public void publishChangeConfig(Map<String, String> changeConfigs) {
			if (changeConfigs.isEmpty()) return;
			if (SYNC_TYPE_ZK.equals(syncType)) {
				if (zkClient == null) {
					logger.warn("Zookeeper client not init,skip");
					return;
				}
				if (zkClient.countChildren(zkPath) == 0) {
					return;
				}
				zkClient.writeData(zkPath, JSON.toJSONString(changeConfigs));
			} else {
				waitingSyncConfigs.putAll(changeConfigs);
			}
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			if (StringUtils.isNotBlank(serverip) && serverport > 0) {
				result = prime * result + ((serverip == null) ? 0 : serverip.hashCode());
				result = prime * result + serverport;
			} else {
				result = prime * result + ((nodeId == null) ? 0 : nodeId.hashCode());
			}
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			ConfigState other = (ConfigState) obj;
			if (StringUtils.isNotBlank(serverip) && serverport > 0) {
				if (!serverip.equals(other.serverip)) return false;
				if (serverport != other.serverport) return false;
			} else {
				if (!nodeId.equals(other.nodeId)) return false;
			}
			return true;
		}
	}
}